"""
Case Type   : 数据库安全特性
Case Name   : 检测异常的无条件删除风险日志
Create At   : 2025/9
Owner       : l1azzzy
Description :
    1、设置enable_risky_query_detection为on
    2、创建测试表并插入测试数据
    3、执行无条件删除语句
    4、检查数据库日志中风险检测记录
Expect      :
    1、参数设置成功
    2、建表和插入数据成功
    3、删除语句执行完成
    4、数据库日志中存在risk关键字及相关无条件delete SQL
History     :
"""

import os
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

class SecurityConfigCase0009(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.com = Common('PrimaryDbUser')
        self.comsh = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.log_path = self.get_log_directory()
        self.test_table = 'test_unconditional_delete'
        self.pri_user = Node(node='PrimaryDbUser')

    def get_log_directory(self):
        """通过SQL查询获取日志目录路径"""
        self.log.info('-----获取日志目录路径-----')
        sql = "show log_directory;"
        
        res = self.comsh.execut_db_sql(sql)
        self.log.info(f"日志目录查询结果: {res}")
        
        lines = res.split('\n')
        for line in lines:
            line = line.strip()
            if not line or line.startswith(('log_directory', '---', '(', 'show ')):
                continue
            if '/' in line:
                self.log.info(f"提取到的日志路径: {line}")
                return line
    
        self.log.info(f"使用默认日志路径: {macro.PG_LOG_PATH}")
        return macro.PG_LOG_PATH
        
    def test_risky_unconditional_delete_detection(self):
        text = '-----step1:设置enable_risky_query_detection为on-----'
        self.log.info(text)
        result = self.comsh.execute_gsguc('reload', 
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_risky_query_detection=on')
        self.assertTrue(result)

        text = '-----step2:创建测试表并插入测试数据-----'
        self.log.info(text)
        create_sql = f'create table {self.test_table} (id int, name text, age int, address text);'
        res = self.comsh.execut_db_sql(create_sql)
        self.log.info(res)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, res)
        insert_sql = f'''
        insert into {self.test_table} 
        select generate_series(1,300), 
            'user_' || generate_series(1,300), 
            (random()*80 + 18)::int,
            'address_' || (random()*100)::int;
        '''
        res = self.comsh.execut_db_sql(insert_sql)
        self.log.info(res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, res)

        text = '-----step3:执行无条件删除语句-----'
        self.log.info(text)
        risky_sqls = [
            f"delete from {self.test_table};",
            f"delete from {self.test_table} where 1=1;",
            f"delete from {self.test_table} where true;",
            f"delete from {self.test_table} where id is not null;"
        ]
        
        for sql in risky_sqls:
            try:
                res = self.comsh.execut_db_sql(sql)
                self.log.info(f"SQL执行结果: {res}")
                if 'DELETE' in res:
                    reinsert_sql = f'''
                    insert into {self.test_table} 
                    select generate_series(1,100), 
                        'user_' || generate_series(1,100), 
                        (random()*80 + 18)::int,
                        'address_' || (random()*100)::int;
                    '''
                    self.comsh.execut_db_sql(reinsert_sql)
            except Exception as e:
                self.log.info(f"SQL执行异常: {e}")
        time.sleep(3)

        text = '-----step4:检查日志中风险检测记录-----'
        self.log.info(text)
        risk_found = False
        self.log.info("遍历所有日志文件查找...")
        cmd = f'find {self.log_path} -name "*.log" -type f'
        res = self.pri_user.sh(cmd).result()
        self.log.info(f"找到的日志文件数量: {len(res.splitlines())}")
        if res.strip():
            log_files = [f.strip() for f in res.split('\n') if f.strip()]
            for log_file in log_files:
                grep_cmd = f'grep -l "Risky.*delete" {log_file} 2>/dev/null'
                grep_res = self.pri_user.sh(grep_cmd).result()
                if grep_res.strip():
                    self.log.info(f"在日志文件 {log_file} 中找到风险记录")
                    detail_cmd = f'grep -i "Risky.delete" {log_file} | head -3'
                    detail_res = self.pri_user.sh(detail_cmd).result()
                    self.log.info(f"风险日志内容: {detail_res}")
                    risk_found = True

        self.assertTrue(risk_found, "未在日志中找到风险记录")

    def tearDown(self):
        text = '-----恢复enable_risky_query_detection为off-----'
        self.log.info(text)
        result = self.comsh.execute_gsguc('reload',
                            self.constant.GSGUC_SUCCESS_MSG,
                            'enable_risky_query_detection=off')
        cleanup_sql = f'drop table if exists {self.test_table};'
        self.comsh.execut_db_sql(cleanup_sql)
        
        self.log.info(f'----{os.path.basename(__file__)} end----')